package io.kimmking.common.rocketmq.repository;

import io.kimmking.dubbo.common.unit.IdGenertor;
import io.kimmking.dubbo.common.unit.ThrowableActionV2;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 通过rocketMQTemplate对生产者各种操作的封装,可以按需进行扩展
 */
@Slf4j
@ConditionalOnBean(RocketMQTemplate.class)
@Component
@RequiredArgsConstructor(onConstructor = @__(@Autowired) )
public class ProducerWrapper<T> {

    private final RocketMQTemplate rocketMQTemplate;
    private final IdGenertor.SnowFlake snowFlake;

    /**
     * 同步方式发送带tag的消息
     * @param topic
     * @param tag
     * @param message
     * @return
     */
    public SendResult sync_send_with_tag(final String topic, final String tag,T message){
        return try_sync_send_with_tag(topic, tag, message, 0l);
    }

    public SendResult try_sync_send_with_tag(final String topic, final String tag, T message, final long timeout){
        return rocketMQTemplate.syncSend(topic+":"+tag, message, timeout);
    }


    /**
     * 设置延时消费
     * @param topic
     * @param message
     * @param timeout
     * @param delayLevel 预设值的延迟时间间隔为：1s、 5s、 10s、 30s、 1m、 2m、 3m、 4m、 5m、 6m、 7m、 8m、 9m、 10m、 20m、 30m、 1h、 2h
     * @return
     */
    public SendResult sync_send_with_delay(String topic, Message<T> message, long timeout, final Integer delayLevel){
        return rocketMQTemplate.syncSend(topic, message, timeout, delayLevel);
    }

    /**
     * 分布式事务性质的发送
     * @param topic
     * @param paramMap 用于构建消息体
     * @param object 业务对象，用于处理本地业务
     * 这里使用snowFlake雪花算法构建事务ID
     * @return
     */
    public TransactionSendResult send_in_tx(String topic, Map<String, Object> paramMap, T object) {

        // 使用map作为消息类型，实际可以自定义个pojo作为消息体
        Message<Map<String, Object>> message = MessageBuilder
                .withPayload(paramMap)
                .setHeader(RocketMQHeaders.PREFIX+RocketMQHeaders.TRANSACTION_ID, snowFlake.nextId())
                .build();
        AtomicReference<TransactionSendResult> transactionSendResult = new AtomicReference();
        try {
            ThrowableActionV2.execute(() -> {
                transactionSendResult.set(rocketMQTemplate.sendMessageInTransaction(topic, message, object));
            });
        } catch (RuntimeException e) {
            // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理。
        }
        return transactionSendResult.get();
    }




}
